﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Timers;
using Lucene.Net.Index;
using Microsoft.Isam.Esent.Collections.Generic;
using Microsoft.TeamFoundation.VersionControl.Client;
using NLog;
using RealTimer = System.Timers.Timer;

namespace TfsGiggle.Core
{
    /// <summary>
    /// Manages jobs in a persisted Dictionary
    /// </summary>
    public sealed class JobManager : IDisposable
    {
        static readonly LoggerWithEventId _logger = (LoggerWithEventId)LogManager.GetCurrentClassLogger(typeof(LoggerWithEventId));
        static readonly Lazy<JobManager> _instance = new Lazy<JobManager>(() => new JobManager());
        static PersistentDictionary<string, Job> _jobs;
        static IndexModel _indexModel;
        CancellationToken _cancelToken;
        readonly RealTimer _jobInspectorTimer = new RealTimer(TimeSpan.FromHours(1).TotalMilliseconds);
        readonly BlockingCollection<Job> _changesetQueue = new BlockingCollection<Job>();
        string _currentRunningJob;

        /// <summary>
        /// My instance
        /// </summary>
        public static JobManager Instance { get { return _instance.Value; } }

        /// <summary>
        /// My IndexModel
        /// </summary>
        public IndexModel IndexModel { get { return _indexModel; } }

        private JobManager()
        {
        }

        /// <summary>
        /// Cleans the job persistent store on regular intervals
        /// </summary>
        /// <param name="state"></param>
        /// <param name="args"></param>
        private void OnInspectJobs(object state, ElapsedEventArgs args)
        {
            _logger.Trace("enter");

            if (_cancelToken.IsCancellationRequested)
            {
                _logger.Trace("Cancellation requested");
                return;
            }

            // Cleanup
            var jobsTooOld = from p in _jobs.Values
                             where p.SubmittedDate.Value < DateTime.UtcNow.AddDays(-Settings.MaxJobAgeInDays)
                             select p;

            _logger.Trace("Removing {0} jobs that are older than {1} days", jobsTooOld.Count(), Settings.MaxJobAgeInDays);
            foreach (var job in jobsTooOld)
            {
                _logger.Trace("Removing job:{0}", job.ID);
                _jobs.Remove(job.ID);
            }

            if (_cancelToken.IsCancellationRequested)
            {
                _logger.Trace("Cancellation requested");
                return;
            }

            var jobsWithTooManyRetries = from p in _jobs.Values where p.RetryCount > Settings.MaxJobRetries select p;
            _logger.Trace("Removing {0} jobs with retries greater than {1}", jobsTooOld.Count(), Settings.MaxJobRetries);
            foreach (var job in jobsWithTooManyRetries)
            {
                _logger.Trace("Removing job:{0}", job.ID);
                _jobs.Remove(job.ID);
            }
            _jobs.Flush();

            var jobsToRestart = from p in JobsWithState(State.Running).
                                    Union(JobsWithState(State.Waiting)).
                                    Union(JobsWithState(State.Error))
                                select p;

            // Restart jobs that may have failed, except any currently running job (single instance ~ single job)
            foreach (var job in jobsToRestart)
            {
                if (job.ID != _currentRunningJob)
                {
                    AddJob(job);
                }
            }
            _logger.Trace("exit");
        }

        private void RestartJobs()
        {
            _logger.Trace("enter");
            _jobInspectorTimer.AutoReset = true;
            _jobInspectorTimer.Elapsed += new ElapsedEventHandler(OnInspectJobs);
            _jobInspectorTimer.Enabled = true;
            _jobInspectorTimer.Start();

            var statusGroups = from p in _jobs.Values group p by p.Status into myGroup select new { myGroup.Key, StatusCount = myGroup.Count() };
            _logger.Trace("Status from persistent jobs:");
            foreach (var group in statusGroups)
            {
                _logger.Trace("   {0} Count:{1}", group.Key, group.StatusCount);
            }

            var jobsToRestart = from p in JobsWithState(State.Running).
                                    Union(JobsWithState(State.Waiting)).
                                    Union(JobsWithState(State.Error))
                                select p;

            // Restart previous jobs eligible for restart
            foreach (var job in jobsToRestart)
            {
                AddJob(job);
            }
            _logger.Trace("exit");
        }

        /// <summary>
        /// Waits for jobs in the blocking queue to ensure one at a time changeset processing.
        /// </summary>
        private void ProcessChangesetQueue()
        {
            _logger.Trace("enter");
            foreach (var job in _changesetQueue.GetConsumingEnumerable())
            {
                _logger.Trace("Processing job: {0} {1}", job.TeamProjectCollectionId, job.ChangesetId);
                ExpandChangeset(job);
            }
        }

        /// <summary>
        /// Set status for this job to the given status
        /// </summary>
        /// <param name="id"></param>
        /// <param name="newStatus"></param>
        /// <returns></returns>
        private State SetStatus(string id, State newStatus)
        {
            return SetStatus(id, newStatus, null);
        }

        private State SetStatus(string id, State newStatus, string message)
        {
            _logger.Trace("id:{0} newStatus:{1} message:{2}", id, newStatus, message);
            if (string.IsNullOrWhiteSpace(id))
            {
                _logger.Trace("Job id is null or empty");
                return State.None;
            }

            if (_jobs.ContainsKey(id))
            {
                var job = _jobs[id];
                var oldStatus = job.Status;
                if (oldStatus == State.Running && newStatus == State.Running)
                {
                    _logger.Warn("Job:{0} seems to have been previously scheduled. Submitted:{1} Changed:{2}", job.ID, job.SubmittedDate, job.ChangedDate);
                }
                job.Status = newStatus;
                job.ChangedDate = DateTime.UtcNow;
                job.ErrorMessage = message;
                job.RetryCount++;
                _jobs[id] = job;
                _jobs.Flush();
                UpdateJobStatistics();
                return oldStatus;
            }
            else
            {
                _logger.Error("Job with id:{0} is not in the job-queue", id);
            }
            _logger.Trace("exit");
            return State.None;
        }

        private void UpdateJobStatistics()
        {
            _logger.Trace("enter");
            Statistics.Instance.JobsRunning.Set(JobsWithStateCount(State.Running));
            Statistics.Instance.JobsError.Set(JobsWithStateCount(State.Error));
            Statistics.Instance.JobsSuccess.Set(JobsWithStateCount(State.Success));
            Statistics.Instance.JobsWaiting.Set(JobsWithStateCount(State.Waiting));
            _logger.Trace("exit");
        }


        /// <summary>
        /// Get status for this task
        /// </summary>
        /// <param name="id"></param>
        /// <returns></returns>
        private State GetStatus(string id)
        {
            _logger.Trace("id:{0}", id);
            if (_jobs.ContainsKey(id))
            {
                return _jobs[id].Status;
            }
            return State.None;
        }

        /// <summary>
        /// Get tasks with a specific state
        /// </summary>
        /// <param name="state"></param>
        /// <returns></returns>
        private Job[] JobsWithState(State state)
        {
            return (from p in _jobs.Values where p.Status == state select p).ToArray();
        }

        /// <summary>
        /// Get #tasks with a specific state
        /// </summary>
        /// <param name="state"></param>
        /// <returns></returns>
        private int JobsWithStateCount(State state)
        {
            return (from p in _jobs.Values where p.Status == state select p).Count();
        }

        private void ExpandChangeset(Job job)
        {
            _logger.Trace("task:{0}", job.ID);
            if (_cancelToken.IsCancellationRequested)
            {
                _logger.Trace("Cancellation requested");
                return;
            }
            SetStatus(job.ID, State.Running);
            _currentRunningJob = job.ID;

            var vcs = TfsSourceControl.Instance.GetVersionControlServer(job.TeamProjectCollectionId);
            if (vcs == null)
            {
                _logger.Warn("Tfs services are currently unavailable");
                return;
            }
            if (_cancelToken.IsCancellationRequested)
            {
                _logger.Trace("Cancellation requested");
                return;
            }
            var processFileTasks = new List<Task>();
            try
            {
                var timer = new Stopwatch();
                timer.Start();

                // Scheduling too many tasks/threads against TFS is bad and will result in timeouts (and still probably can)
                var lcts = new LimitedConcurrencyLevelTaskScheduler(Settings.MaxNumberOfTasksToProcessChangeset);
                var factory = new TaskFactory(lcts);
                using (var writer = IndexModel.GetIndexWriter())
                {
                    var changeset = vcs.GetChangeset(job.ChangesetId, true, false);
                    _logger.Trace("There seems to be {0} changes in changeset {1}", changeset.Changes.Length, job.ChangesetId);
                    int numberOfAddOrDeletions = 0;
                    foreach (var change in changeset.Changes)
                    {
                        if (_cancelToken.IsCancellationRequested)
                        {
                            _logger.Trace("Cancellation requested");
                            return;
                        }
                        if (!IndexModel.IsSupportedFileExtension(change.Item.ServerItem))
                        {
                            _logger.Trace("Filetype {0} is not indexed. Add the extension to the RegexForKnownFileExtensions setting in the config to include it", change.Item.ServerItem);
                            continue;
                        }
                        processFileTasks.Add(CreateProcessFileTask(factory, writer, job.TeamProjectCollectionId, change));
                        numberOfAddOrDeletions++;
                    }
                    var childTasks = processFileTasks.ToArray();
                    _logger.Trace("Waiting for {0} index tasks", childTasks.Length);

                    // Wait for all tasks to complete, so that we can complete the job
                    Task.WaitAll(childTasks);
                    SetStatus(job.ID, State.Success);
                    Statistics.Instance.NumberOfChangesetsExpanded.Add(1);
                    IndexModel.Commit(writer);
                    timer.Stop();
                    _logger.Trace("Added/Deleted {0} files in {1} seconds with {2} tasks", numberOfAddOrDeletions, timer.ElapsedMilliseconds / 1000, childTasks.Length);
                }
            }
            catch (Exception ex)
            {
                _logger.ErrorExceptionWithEventId(EventId.ExpandChangeset, "exception caught", ex);
                SetStatus(job.ID, State.Error, ex.ToString());
            }
            finally 
            {
                _currentRunningJob = null;
            }
            _logger.Trace("exit");
        }

        private Task CreateProcessFileTask(TaskFactory factory, IndexWriter writer, string teamProjectCollectionId, IChange change)
        {
            _logger.Trace("enter");
            if ((change.ChangeType & ChangeType.Delete) == ChangeType.Delete)
            {
                return factory.StartNew(() => _indexModel.Delete(writer, teamProjectCollectionId, change.Item.ServerItem));
            }
            else
            {
                return factory.StartNew(() => _indexModel.Add(writer, teamProjectCollectionId, change.Item.ServerItem));
            }
        }

        private void Dispose(bool disposing)
        {
            _logger.Trace("disposing");
            if (disposing)
            {
                if (_indexModel != null)
                {
                    _indexModel.Dispose();
                }
                if (_changesetQueue != null)
                {
                    _changesetQueue.Dispose();
                }
                if (_jobInspectorTimer != null)
                {
                    _jobInspectorTimer.Dispose();
                }
                if (_jobs != null)
                {
                    _jobs.Dispose();
                }
            }
        }

        /// <summary>
        /// Add job to the blocking queue to ensure only one processing changeset at a time
        /// </summary>
        /// <param name="job"></param>
        internal void AddJob(Job job)
        {
            _logger.Trace("{0} {1}", job.TeamProjectCollectionId, job.ChangesetId);
            _changesetQueue.Add(job);
        }

        /// <summary>
        /// Adds a changeset to the blocking queue to ensure only one processing changeset at a time
        /// </summary>
        /// <param name="teamProjectCollectionId"></param>
        /// <param name="changesetId"></param>
        public void AddJob(string teamProjectCollectionId, int changesetId)
        {
            _logger.Trace("teamProjectCollectionId:{0} changesetId:{1}", teamProjectCollectionId, changesetId);
            var job = new Job(teamProjectCollectionId, changesetId);
            _jobs.Add(job.ID, job);
            _jobs.Flush();
            AddJob(job);
            _logger.Trace("exit");
        }

        /// <summary>
        /// Initialize dependencies
        /// </summary>
        /// <param name="dataDirectory"></param>
        /// <param name="tfsConfigurationServer"></param>
        /// <param name="cancelToken"></param>
        public void Open(string dataDirectory, ITfsConfigurationServer tfsConfigurationServer, CancellationToken cancelToken)
        {
            if (string.IsNullOrWhiteSpace(dataDirectory)) throw new ArgumentNullException("dataDirectory");
            if (tfsConfigurationServer == null) throw new ArgumentNullException("tfsConfigurationServer");

            _logger.Trace("dataDirectory:{0} tfsUri:{1}", dataDirectory, tfsConfigurationServer);
            try
            {
                // TODO Microsoft.Isam.Esent.Interop.EsentSystemPathInUseException was unhandled by user code
                //  Message=System path already used by another database instance Source=Esent.Interop
                _jobs = new PersistentDictionary<string, Job>(Path.Combine(dataDirectory, "Jobs"));
                _logger.Trace("{0} persisted jobs in Jobs", _jobs.Count);
                _indexModel = new IndexModel(Path.Combine(dataDirectory, "Index"), new MultiLanguageAnalyzer(), tfsConfigurationServer, cancelToken);
                Statistics.Instance.Open(Path.Combine(dataDirectory, "Statistics"));
                Statistics.Instance.NumberOfDocuments.Set(IndexModel.NumberOfDocuments);
                _cancelToken = cancelToken;
                Task.Factory.StartNew(() => ProcessChangesetQueue());
                RestartJobs();
            }
            catch (Exception ex)
            {
                _logger.ErrorExceptionWithEventId(EventId.JobManagerOpen, "Exception caught", ex);
                throw;
            }
            _logger.Trace("exit");
        }

        /// <summary>
        /// Go away
        /// </summary>
        public void Dispose()
        {
            _logger.Trace("enter");
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        /// Return TfsGiggle statistics
        /// </summary>
        /// <returns></returns>
        public string GetStatistics()
        {
            return Statistics.Instance.ToString();
        }
    }
}